package com.realife.eson.common.util.mq;


import com.realife.eson.common.util.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * @version V1.0
 * @author: guangcheng
 * @date: 2022/5/31 21:37
 * @description:
 */
@Slf4j
@Component
@SuppressWarnings("all")
public class KafkaUtil {

    //自定义topic     test-consumer-group
    public static final String TOPIC_GIT_CREATE_PROJECT = "topic-git-create-project";
    public static final String TOPIC_ARGOCD_CREATE_APPLICATION = "topic-argocd-create-application";

    //group
    public static final String GROUP_ARGOCD_CREATE_APPLICATION = "addargoapp";
    public static final String GROUP_DEFAULT = "";

    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    IdWorker idWorker;


    /**
     * @param topic
     * @param msg
     */
    public void sendMsg(String topic, String msg) {
        //String nextId = UniqueCodeUtil.nextId();
        /*ListenableFuture<SendResult<String, Object>> future = */
        kafkaTemplate.send(topic, msg);
//        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
//            @Override
//            public void onFailure(Throwable throwable) {
//                log.info("kafka producer send message error: topic={}, error={}\n", topic, throwable.getMessage());
//            }
//
//            @Override
//            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
//                log.info("kafka producer send msg ok, topic={}, msg={}\n", topic, stringObjectSendResult.toString());
//            }
//        });
    }


//    /**
//     *
//     * @param record
//     * @param ack
//     * @param topic
//     */
//    @KafkaListener(topics = TOPIC_TEST, groupId = TOPIC_GROUP1)
//    //@KafkaHandler
//    public void consumerMsg1(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
//        Optional<?> message = Optional.ofNullable(record.value());
//        if (message.isPresent()) {
//            Object msg = message.get();
//            log.info("kafka consumer msg: topic={}, message={}\n", topic, msg);
//            ack.acknowledge();
//        }
//    }


//    /**
//     *
//     * @param record
//     * @param ack
//     * @param topic
//     */
//    @KafkaListener(topics = TOPIC_TEST, groupId = TOPIC_GROUP2)
//    public void consumerMsg2(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
//        Optional message = Optional.ofNullable(record.value());
//        if (message.isPresent()) {
//            Object msg = message.get();
//            log.info("kafka consumer msg: topic={}, message={}\n", topic, msg);
//            ack.acknowledge();
//        }
//    }


//    @KafkaListener(topics = {TOPIC_ARGOCD_CREATE_APPLICATION}, groupId = "")
//    public void onMessage(String msg) {
//        System.out.println("msg =\n" + msg);
//    }
//
//
//    @KafkaListener(topics = {TOPIC_ARGOCD_CREATE_APPLICATION}, groupId = "")
//    public void listener(ConsumerRecord record) {
//        Object value = record.value();
//        if (AssertNP.legalParams(value)) {
//            System.out.println("value =\n" + value);
//        }
//    }
}

